<?php
/**
 * 队列模块: 包含生产队列
 * Created by PhpStorm.
 * User: liujianlin
 * Date: 2017/5/18
 * Time: 18:26
 */

namespace app\controllers\marketing;

use yii;
use yii\web\Controller;
use app\models\marketing\MailGroup;
use app\models\marketing\MailSubgroup;
use app\models\marketing\MailSendlist;
use app\models\marketing\MailTemplate;
use app\components\redis\RedisMarketingEmail;
use app\components\channel\Channel;
use app\components\FileLock;
use app\models\marketing\User;
use app\components\redis\RedisLock;
use app\models\marketing\GroupOperationLog;
use app\components\mq\MqComponent;
//use app\controllers\marketing\QueueController;
use app\models\marketing\Queue;
use app\components\msgcenter\MsgCenter;
use app\models\Elog;
use Ares333\Curl\Toolkit;
use app\components\SplitSendlist;

class QueueController extends Controller
{
    /**
     * redis连接
     * @var RedisMarketingEmail
     */
    private $redis;

    /**
     * MailGroup对象，用于操作分组model
     * @var MailGroup
     */
    private $groupModel;

    /**
     * MailSubgroup对象，用于操作子分组model
     * @var MailSubgroup
     */
    private $subgroupModel;

    /**
     * MailSendlist对象，用于操作发送列表model
     * @var MailSendlist
     */
    private $sendListModel;

    /**
     * 开始时间
     * @var int
     */
    private $startTime;

    /**
     * 配置信息
     * @var array
     */
    private $config;

    /**
     * @var bool whether to enable CSRF validation for the actions in this controller.
     * CSRF validation is enabled only when both this property and [[\yii\web\Request::enableCsrfValidation]] are true.
     */
    public $enableCsrfValidation = false;

    /**
     * 初始化
     */
    public function init()
    {
        $this->redis = new RedisMarketingEmail();
        $this->groupModel = new MailGroup();
        $this->subgroupModel = new MailSubgroup();
        $this->sendListModel = new MailSendlist();
        $this->startTime = time();
        
        // $config['errorCode'] = [7001,6001];                 // 通道商返回这些错误码时不再马上发送和重发
        $config['errorCode'] = [403];                 // 通道商返回这些错误码时不再马上发送和重发
        $config['sendSpacingTime'] = 3600;                  // 返回上面错误码时重发的间隔时间
        $config['maxSubgroupFailCount'] = 5;                // 子分组最大失败(用于重发)次数
        $config['maxGroupFailCount'] = 20;                  // 分组最大失败(用于重发，子分组失败1次记一次)次数
        $config['maxResendProcessTime'] = 55;               // 重发进程处理时间（单位：秒）
        $config['maxProcedureProcessTime'] = 55;            // 生产进程处理时间
        $config['maxWorkerProcessTime'] = 50;               // worker进程处理时间
        $config['maxRedisMemory'] = 10 * 1024 * 1024 * 1024;     // 10G
        $config['stopPriority'] = 10000;                    // 暂停分组的优先级
        $config['maxSuspendLen'] = 1000;            // 最大暂停子分组的数量
        $config['maxSuspendTime'] = 3600;           // 最大暂停时间（单位：秒）
        $config['businessRedoTimes'] = 5;           // 各业务失败后重试次数 
        // 消费master相关配置
        $config['machineNum'] = 2;              // 机器数量
        $config['workerNum'] = 8;               // 每台机器的worker数
        $config['sendSpeed'] = 20;              // 每分钟发送的子分组数
        $config['workerMaxAliveTime'] = 300;    // worker的最大存活时间，单位：秒
        $config['onceCutStep2GroupNums'] = 5;   // 每次生成缓存的分组数为5 
        $config['cookieArr'] = [
            1 => 'ORIGINDC=1;Domain=.appinthestore.com;Path=/',
            2 => 'ORIGINDC=2;Domain=.appinthestore.com;Path=/'
        ];        // 负载均衡的cookie设置

        $workerNum = $this->redis->hget('emarsys_new_simulate', 'worker_num');
        if (!empty($workerNum)) {
            $config['workerNum'] = $workerNum * 1;
        }
        $this->config = $config;
        yii::$app->db->enableSlaves = false;        // 禁用从库查询，从主库查询数据。防止生成子分组时主从不同步
    }
    
    /**
     * 生产队列rabbitmq版（sendlist拆分，切割发送部分优化）
     */
    public function actionProcedureGroupAllRabbitmq()
    {
        header("Content-type:text/html;charset=utf-8");
        $this->writeProcedureLog("开始进入生产队列rabbitmq版");
        echo "<title>所有分组都进入此生产队列rabbitmq版</title>";
        if (!empty($_GET['group_id'])) {
            $groupArr = $this->groupModel->getAllPendingPushGroup($_GET['group_id']);           //  把相关的分组id放到rabbitmq中
        } else {
            $groupArr = $this->groupModel->getAllPendingPushGroup();           //  把相关的分组id放到rabbitmq中
        }
        
        // 往mq中加入数据
        if (!empty($groupArr) && is_array($groupArr)) {
            Elog::logCrontab(json_encode($groupArr),__FUNCTION__,'info');
            foreach ($groupArr as $groupArrVal) {
                try {
                    $trans = Yii::$app->db->beginTransaction(); // 开启事务
                    $updateResult = $this->groupModel->updateGroupsByGroupIdsIsInRabbitmqQueue($groupArrVal['group_id'],1);
                    if (!empty($updateResult)) {
                       
                            $returnFlag = MqComponent::sendMqData('sub_all_group_step_EMP',json_encode($groupArrVal));
                            if ($returnFlag) {

                                // mysql数据提交
                                $trans->commit();
                            } else { // 说明加入到mq中失败了
                                // 改成抛异常 2-7 
                                throw new \Exception("当前分组加入mq中失败");
                            }
                    }
                } catch (\Exception $ex) {  // 其他异常
               
                    // mysql数据回滚
                    $trans->rollBack();
                    $msg = __FUNCTION__."方法，分组{$groupArrVal['group_id']}生产过程出现问题.具体错误信息为:" . $ex->getMessage();
                    echo $msg;
                    Elog::logCrontab($msg, __FUNCTION__,'exception'); // 记录信息到elog
                    // 发送信息到消息中心
                    // 发送错误消息
                    $msgCenter = new MsgCenter();
                    $msgCenter->sendSubgroupProductMessage($msg,$msg);  
                }
            }
            echo '分组成功加入队列';
        } else {
            echo '没有适合条件的分组';
        }
    }
    
    /**
     * 生产队列rabbitmq版(分组状态为推送)
     */
    public function actionProcedurePushGroupRabbitmq()
    {
        header("Content-type:text/html;charset=utf-8");
        $this->writeProcedureLog("开始进入生产队列rabbitmq版");
        echo "<title>生产队列rabbitmq版</title>";
        if (!empty($_GET['group_id'])) {
            $groupArr = $this->groupModel->getPendingPushGroup($_GET['group_id']);           //  把相关的分组id放到rabbitmq中
        } else {
            $groupArr = $this->groupModel->getPendingPushGroup();           //  把相关的分组id放到rabbitmq中
        }
        
        // 往mq中加入数据
        if (!empty($groupArr) && is_array($groupArr)) {
            Elog::logCrontab(json_encode($groupArr), 'actionProcedureGroupRabbitmq','info');
            foreach ($groupArr as $groupArrVal) {
                try {
                    $trans = Yii::$app->db->beginTransaction(); // 开启事务
                    $updateResult = $this->groupModel->updateGroupsByGroupIdsIsInRabbitmqQueue($groupArrVal['group_id'],1);
                    if (!empty($updateResult)) {
                       
                            $returnFlag = MqComponent::sendMqData('sub_group_step_EMP',json_encode($groupArrVal));
                            if ($returnFlag) {

                                // mysql数据提交
                                $trans->commit();
                            } else { // 说明加入到mq中失败了
                                // 改成抛异常 2-7 
                                throw new \Exception("当前分组加入mq中失败");
                            }
                    }
                } catch (\Exception $ex) {  // 其他异常
               
                    // mysql数据回滚
                    $trans->rollBack();
                    $msg = "actionProcedurePushGroupRabbitmq方法，分组{$groupArrVal['group_id']}生产过程出现问题.具体错误信息为:" . $ex->getMessage();
                    echo $msg;
                    Elog::logCrontab($msg, 'actionProcedurePushGroupRabbitmq','exception'); // 记录信息到elog
                    // 发送信息到消息中心
                    // 发送错误消息
                    $msgCenter = new MsgCenter();
                    $msgCenter->sendSubgroupProductMessage($msg,$msg);  
                }
                
            }
            echo '分组成功加入队列';
        } else {
            echo '没有适合条件的分组';
        }
    }

    /**
     * 生产队列rabbitmq版
     */
    public function actionProcedureGroupRabbitmq()
    {
        header("Content-type:text/html;charset=utf-8");
        $this->writeProcedureLog("开始进入生产队列rabbitmq版");
        echo "<title>生产队列rabbitmq版</title>";
        $groupArr = $this->groupModel->getPendingGroup();           //  把相关的分组id放到rabbitmq中
        
        // 往mq中加入数据
        if (!empty($groupArr) && is_array($groupArr)) {
            Elog::logCrontab(json_encode($groupArr), 'actionProcedureGroupRabbitmq','info');
            foreach ($groupArr as $groupArrVal) {
                $returnFlag = MqComponent::sendMqData('sub_group_EMP',json_encode($groupArrVal));
                if (!empty($returnFlag)) { // 说明插入队列成功，更改数据库中状态
                    $this->groupModel->updateGroup($groupArrVal['group_id'],['is_in_rabbitmq_queue'=>1]);
                }
            }
            echo '分组成功加入队列';
        } else {
            echo '没有适合条件的分组';
        }
    }
    
    /**
     * 重发队列rabbitmq版 拆表后专用
     */
    public function actionProcedureAllGroupResendRabbitmq()
    {
        // 这里也要加事务 2-7
        header("Content-type:text/html;charset=utf-8");
        $this->writeProcedureLog("开始进入生产队列rabbitmq版");
        echo "<title>生产队列rabbitmq版</title>";
        $groupArr = $this->groupModel->getResendPushAllCroup();           //  把相关的分组id放到rabbitmq中
        
        // 往mq中加入数据
        if (!empty($groupArr) && is_array($groupArr)) {
            Elog::logCrontab(json_encode($groupArr), 'actionProcedureGroupResendRabbitmq','info');
            foreach ($groupArr as $groupArrVal) {
                try {
                    $trans = Yii::$app->db->beginTransaction(); // 开启事务
                    $updateResult = $this->groupModel->updateGroupsByGroupIdsIsInRabbitmqResendQueue($groupArrVal['group_id'],1);
                    if (!empty($updateResult)) {
                        $returnFlag = MqComponent::sendMqData('sub_all_group_resend_EMP',json_encode($groupArrVal));
                        if ($returnFlag) {

                            // mysql数据提交
                            $trans->commit();
                        } else { // 说明加入到mq中失败了
                            // 改成抛异常 2-7 
                            throw new \Exception("当前分组加入mq中失败");
                        }
                    }
                } catch (\Exception $ex) {
                        
                    // mysql数据回滚
                    $trans->rollBack();
                    $msg = "actionProcedureGroupResendRabbitmq方法，分组{$groupArrVal['group_id']}生产过程出现问题.具体错误信息为:" . $ex->getMessage();
                    echo $msg;
                    Elog::logCrontab($msg, 'actionProcedureGroupResendRabbitmq','exception'); // 记录信息到elog
                    // 发送信息到消息中心
                    // 发送错误消息
                    $msgCenter = new MsgCenter();
                    $msgCenter->sendSubgroupProductMessage($msg,$msg);  
                }
            }
            echo '分组成功加入队列';
        } else {
            echo '没有适合条件的分组';
        }
    }
    /**
     * 生产队列rabbitmq版
     */
    public function actionProcedureGroupResendRabbitmq()
    {
        // 这里也要加事务 2-7
        header("Content-type:text/html;charset=utf-8");
        $this->writeProcedureLog("开始进入生产队列rabbitmq版");
        echo "<title>生产队列rabbitmq版</title>";
        $groupArr = $this->groupModel->getResendPushGroup();           //  把相关的分组id放到rabbitmq中
        
        // 往mq中加入数据
        if (!empty($groupArr) && is_array($groupArr)) {
            Elog::logCrontab(json_encode($groupArr), 'actionProcedureGroupResendRabbitmq','info');
            foreach ($groupArr as $groupArrVal) {
                try {
                    $trans = Yii::$app->db->beginTransaction(); // 开启事务
                    $updateResult = $this->groupModel->updateGroupsByGroupIdsIsInRabbitmqResendQueue($groupArrVal['group_id'],1);
                    if (!empty($updateResult)) {
                        $returnFlag = MqComponent::sendMqData('sub_group_resend_EMP',json_encode($groupArrVal));
                        if ($returnFlag) {

                            // mysql数据提交
                            $trans->commit();
                        } else { // 说明加入到mq中失败了
                            // 改成抛异常 2-7 
                            throw new \Exception("当前分组加入mq中失败");
                        }
                    }
                } catch (\Exception $ex) {
                        
                    // mysql数据回滚
                    $trans->rollBack();
                    $msg = "actionProcedureGroupResendRabbitmq方法，分组{$groupArrVal['group_id']}生产过程出现问题.具体错误信息为:" . $ex->getMessage();
                    echo $msg;
                    Elog::logCrontab($msg, 'actionProcedureGroupResendRabbitmq','exception'); // 记录信息到elog
                    // 发送信息到消息中心
                    // 发送错误消息
                    $msgCenter = new MsgCenter();
                    $msgCenter->sendSubgroupProductMessage($msg,$msg);  
                }
            }
            echo '分组成功加入队列';
        } else {
            echo '没有适合条件的分组';
        }
    }
    
    
    /**
     * 清空键值
     */
    public function actionDelProcedureProgressNums()
    {
        $this->redis->set('procedure_progress_nums',0,300);
    }
    
    /**
     * 清空键值
     */
    public function actionDelProcedureSubgroupStep1Nums()
    {
        $this->redis->set('procedure_subgroup_step1_nums',0,300);
    }
    
    /**
     * 清空键值
     */
    public function actionDelProcedureSubgroupStep2Nums()
    {
        $this->redis->set('procedure_subgroup_step2_nums',0,300);
    }
    
    /**
     * 清空键值
     */
    public function actionDelProcedureProgressResendNums()
    {
        $this->redis->set('procedure_progress_resend_nums',0,300);
    }
    
    /*
     * 消费rabbitmq中分组数据，计算出当前分组要切分成多少个子分组（比如说1000个子分组），
     * 往mysql子分组表插入数据，往redis队列中插入1000条子分组明细数据（用队列保存，批量插入用另外一个键值保存（因为当前这个子分组还不可以发送）），
     * 成功后应答。
     */
    public function actionProcedureAllSubgroupStep1()
    {
        set_time_limit(0);
        $obj = new Queue;
        MqComponent::receiveQueueData('sub_all_group_step_EMP',$obj,'dealAllGroupRabbitStep1');
    }
    
    /*
     * 消费rabbitmq中分组数据，计算出当前分组要切分成多少个子分组（比如说1000个子分组），
     * 往mysql子分组表插入数据，往redis队列中插入1000条子分组明细数据（用队列保存，批量插入用另外一个键值保存（因为当前这个子分组还不可以发送）），
     * 成功后应答。
     */
    public function actionProcedureSubgroupStep1()
    {
        set_time_limit(0);
        $obj = new Queue;
        MqComponent::receiveQueueData('sub_group_step_EMP',$obj,'dealGroupRabbitStep1');
    }
    
    /**
     * 喻茂
     * 2018-8-25
     * 管理actionProcedureAllSubgroupStep2，可以并发处理
     */
    public function actionProcedureAllSubgroupMasterStep2()
    {
        ignore_user_abort(true);     // 忽略用户的中断
        set_time_limit(180);
        $config = $this->config;
        header("Content-type:text/html;charset=utf-8");
        echo "<title>分割子分步骤二，生成具体子分组缓存数据</title>";
        $redis = new RedisMarketingEmail();
        $redisLock = new RedisLock($redis,__FUNCTION__, 900);
        if ($redisLock->translock() === false) {
            $msg = __FUNCTION__ . "进程已经在运行";
            Elog::logCrontab($msg,__FUNCTION__,'info');
            return 0;
        }
        try {
            // 获取还需要处理的分组
            $subgroupModel = $this->subgroupModel;
            $groupIdArr = $subgroupModel->getMailHaveCutSubgroupAll(); 
            if ($groupIdArr) {
                $toolkit = new Toolkit();
                $curl = $toolkit->getCurl();
                $curl->onInfo = null;
                // 遍历启动多个进程，生成子分组缓存
                foreach ($groupIdArr as $groupId) {
                    $groupId = $groupId['group_id'];
                    $url = "http://{$_SERVER['HTTP_HOST']}/marketing/queue/procedure-all-subgroup-step2"
                    . "?module_name=marketing_email&group_id={$groupId}";
                    $curl->add(
                        array(
                            'opt' => array(
                                CURLOPT_URL => $url,
                            )
                        ),
                        function ($r) {
                            echo "Request success for " . $r['info']['url'] . "\n";
                            echo $r['body']."\n";
                        }); 
                    $msg = "通过访问{$url}处理{$groupId}分组开始";
                    echo $msg;
                    Elog::logCrontab($msg,__FUNCTION__,'info'); // 记录信息到elog
                }
                $curl->start();
            } else {
                $msg = "当前没有所有启动分组state=6的子分组都已经生成缓存";
                echo $msg;
                Elog::logCrontab($msg, 'actionProcedureSubgroupMasterStep2','info'); // 记录信息到elog
            }
        } catch (\Exception $ex) {
            $msg = __FUNCTION__ . "进程出现异常，具体异常信息为:".$ex->getMessage();
            echo $msg;
            Elog::logCrontab($msg, __FUNCTION__,'exception'); // 记录信息到elog
        }
        $redisLock->unlock();
    }
    
    /**
     * 喻茂
     * 2018-3-1
     * 管理actionProcedureSubgroupStep2，可以并发处理
     */
    public function actionProcedureSubgroupMasterStep2()
    {
        ignore_user_abort(true);     // 忽略用户的中断
        set_time_limit(180);
        $config = $this->config;
        header("Content-type:text/html;charset=utf-8");
        echo "<title>分割子分步骤二，生成具体子分组缓存数据</title>";
        $lock = new FileLock("ProcedureSubgroupMasterStep2");
        if (false === $lock->tryLock()) {
            exit("获取文件锁失败，ProcedureSubgroupMasterStep2进程已在运行");
        }
        
        // 获取还需要处理的分组
        $subgroupModel = $this->subgroupModel;
        $groupIdArr = $subgroupModel->getMailHaveCutSubgroup($config['onceCutStep2GroupNums']); 
        if ($groupIdArr) {
            // 遍历启动多个进程，生成子分组缓存
            foreach ($groupIdArr as $groupId) {
                $groupId = $groupId['group_id'];
                $url = "http://{$_SERVER['HTTP_HOST']}/marketing/queue/procedure-subgroup-step2"
                . "?module_name=marketing_email&group_id={$groupId}";
                $item['url'] = $url;
                $item['options'][CURLOPT_POST] = 0;
           
                $item['options'][CURLOPT_RETURNTRANSFER] = 1;
                $item['options'][CURLOPT_TIMEOUT] = 100;
                yii::$app->curl->execute($item);
                $msg = "通过访问{$url}处理{$groupId}分组开始";
                echo $msg;
                Elog::logCrontab($msg, 'actionProcedureSubgroupMasterStep2','info'); // 记录信息到elog
            }
        } else {
            $msg = "当前没有所有启动分组state=6的子分组都已经生成缓存";
            echo $msg;
            Elog::logCrontab($msg, 'actionProcedureSubgroupMasterStep2','info'); // 记录信息到elog
        }
    }
    
    /**
     * 喻茂
     * 2018-8-25
     * 处理mysql中mail_subgroup表中状态为6的数据生成具体的邮箱缓存数据和具体明细数据到缓存中
     */
    public function actionProcedureAllSubgroupStep2()
    {
        set_time_limit(0);
        
        // 具体操作
        $obj = new Queue;
        $returnCheckInfo = $this->procedureLimitCheckMemory();
        if (!empty($returnCheckInfo) && $returnCheckInfo['code'] == 1) { // 说明redis内存溢出
            $msg = __FUNCTION__."方法生成子分组具体缓存信息时出现异常：（{$returnCheckInfo['info']}）";
            echo $msg;
            Elog::logCrontab($msg,__FUNCTION__,'exception'); // 记录信息到elog
            // 发送错误消息
            $msgCenter = new MsgCenter();
            $msgCenter->sendSubgroupProductMessage($msg,$msg);
            exit;
        }
        if (!empty($_GET['group_id'])) {
            // 加锁，只有一个进程在处理同一个group_id
            $groupId = intval($_GET['group_id']);
            $redis = new RedisMarketingEmail();
            $redisLock = new RedisLock($redis, __FUNCTION__ . $groupId, 900);
            if ($redisLock->translock() === false) {
                $msg = __FUNCTION__ . "子分组{$groupId}已经在切割中";
                Elog::logCrontab($msg,__FUNCTION__,'info');
                return 0;
            }
            // 查看每个send_day的子分组数量
            $subgroupModel = $this->subgroupModel;
            $subgroupCoutByDayArrUndeal = $subgroupModel->getSubgroupCoutBySendDay($_GET['group_id']);
            $subgroupCoutByDayArr = [];
            if ($subgroupCoutByDayArrUndeal) {
                foreach ($subgroupCoutByDayArrUndeal as $subgroupCoutByDayArrUndealVal) {
                    $subgroupCoutByDayArr[$subgroupCoutByDayArrUndealVal['send_day']] = $subgroupCoutByDayArrUndealVal['subgroupCountCurDay'];
                }
            }
            
            while(1) {
                $returnCheckInfo = $this->procedureLimitCheckTime();
                if ($returnCheckInfo['code'] == 2) { // 超时
                    echo '当前进程执行超时';
                    break;
                }
                
                $returnInfo = $obj->dealAllGroupRabbitStep2($_GET['group_id'],$subgroupCoutByDayArr);
                if ($returnInfo['code'] != 0) {
                    if ($returnInfo['code'] == 100) {
                        echo $returnInfo['info'];
                        break;
                    } elseif($returnInfo['code'] != 11) {
                        $msg = __FUNCTION__ . "方法，生成子分组具体缓存信息时出现异常：（{$returnInfo['code']}_{$returnInfo['info']}）";
                        echo $msg;
                        Elog::logCrontab($msg, __FUNCTION__,'exception'); // 记录信息到elog
                        // 发送错误消息
                        $msgCenter = new MsgCenter();
                        $msgCenter->sendSubgroupProductMessage($msg,$msg);
                        if ($returnInfo['code'] == 10) {
                            break;
                        }
                    } else {
                        $msg = __FUNCTION__ . "方法，生成子分组具体缓存信息时出现异常：（{$returnInfo['code']}_{$returnInfo['info']}）";
                        echo $msg;
                        Elog::logCrontab($msg, __FUNCTION__,'exception'); // 记录信息到elog
                    } 
                }
            }
            $redisLock->unlock();
        } else {
            echo '必须指定group_id值';
            exit;
        }
    }
    
    /**
     * 喻茂
     * 2018-3-1
     * 处理mysql中mail_subgroup表中状态为6的数据生成具体的邮箱缓存数据和具体明细数据到缓存中，并且绑定mysql队列表（sendlist表）中相关记录
     */
    public function actionProcedureSubgroupStep2()
    {
        set_time_limit(0);
        
        // 具体操作
        $obj = new Queue;
        while (1) {
            $returnCheckInfo = $this->procedureLimitCheckNew();
            if ($returnCheckInfo['code'] == 0) { // 内存没有溢出，没有超时
                
                if (!empty($_GET['group_id'])) { // 可以指定切割某一子分组的数据
                    $returnInfo = $obj->dealGroupRabbitStep2($_GET['group_id']);
                } else {
                    $returnInfo = $obj->dealGroupRabbitStep2();
                }
                
                if ($returnInfo['code'] != 0) {
                    if ($returnInfo['code'] == 100) {
                        echo $returnInfo['info'];
                        break;
                    } elseif($returnInfo['code'] != 11) {
                        $msg = "actionProcedureSubgroupStep2方法，生成子分组具体缓存信息时出现异常：（{$returnInfo['code']}_{$returnInfo['info']}）";
                        echo $msg;
                        Elog::logCrontab($msg, 'actionProcedureSubgroupStep2','exception'); // 记录信息到elog
                        // 发送错误消息
                        $msgCenter = new MsgCenter();
                        $msgCenter->sendSubgroupProductMessage($msg,$msg);
                        if ($returnInfo['code'] == 10) {
                            break;
                        }
                    } else {
                        $msg = "actionProcedureSubgroupStep2方法，生成子分组具体缓存信息时出现异常：（{$returnInfo['code']}_{$returnInfo['info']}）";
                        echo $msg;
                        Elog::logCrontab($msg, 'actionProcedureSubgroupStep2','exception'); // 记录信息到elog
                    } 
                }
            } else { // 说明redis内存溢出或超时
                if ($returnCheckInfo['code'] == 1) { // 内存溢出
                    $msg = "actionProcedureSubgroupStep2方法，生成子分组具体缓存信息时出现异常：（{$returnCheckInfo['info']}）";
                    echo $msg;
                    Elog::logCrontab($msg, 'actionProcedureSubgroupStep2','exception'); // 记录信息到elog
                    // 发送错误消息
                    $msgCenter = new MsgCenter();
                    $msgCenter->sendSubgroupProductMessage($msg,$msg);
                    break;
                } else { // 超时
                    $msg = "actionProcedureSubgroupStep2方法执行超时，不用进行处理，这是正常信息。";
                    echo $msg;
                    Elog::logCrontab($msg, 'actionProcedureSubgroupStep2','info'); // 记录信息到elog
                    break;
                } 
            }
            // sleep(1);
        }  
    }
    
    /**
     * 消费rabbitmq中的相关分组数据生成子分组(sendlist表拆分版)
     */
    public function actionProcedureResendAllSubgroup()
    {
        set_time_limit(0);
        $obj = new Queue;
        MqComponent::receiveQueueData('sub_all_group_resend_EMP',$obj,'dealAllGroupRabbitResend');
      
    }
    
     /**
     * 消费rabbitmq中的相关分组数据生成子分组(新的发送架构重发)
     */
    public function actionProcedureResendSubgroup()
    {
        set_time_limit(0);
        $maxProcedureProgress = 10; // 最大启用10个进程
        
        // 从redis中获取进程数量
        $procedureProgressNums = $this->redis->get('procedure_progress_resend_nums');
        if ($procedureProgressNums===NULL) {
            
            // 5分钟把分组的is_in_rabbitmq_queue恢复成0
            $this->groupModel->updateIsInRabbitmqQueueStatus();
            $this->redis->set('procedure_progress_resend_nums',0,300);
            $procedureProgressNums=0;
        }
        if (intval($procedureProgressNums) >= $maxProcedureProgress) {
            echo '生产进程数超过了最大限制';
            exit;
        } else {
            $this->redis->incr('procedure_progress_resend_nums');
        }
        $obj = new Queue;
        MqComponent::receiveQueueData('sub_group_resend_EMP',$obj,'dealGroupRabbitResend');
        $procedureProgressNums = $this->redis->get('procedure_progress_resend_nums');
        if($procedureProgressNums > 0) {
            $this->redis->decr('procedure_progress_resend_nums');
        }
    }
    /**
     * 消费rabbitmq中的相关分组数据生成子分组
     */
    public function actionProcedureSubgroup()
    {
        set_time_limit(0);
        $maxProcedureProgress = 10; // 最大启用10个进程
        
        // 从redis中获取进程数量
        $procedureProgressNums = $this->redis->get('procedure_progress_nums');
        if ($procedureProgressNums===NULL) {
            
            // 5分钟把分组的is_in_rabbitmq_queue恢复成0
            $this->groupModel->updateIsInRabbitmqQueueStatus();
            $this->redis->set('procedure_progress_nums',0,300);
            $procedureProgressNums=0;
        }
        if (intval($procedureProgressNums) >= $maxProcedureProgress) {
            echo '生产进程数超过了最大限制';
            exit;
        } else {
            $this->redis->incr('procedure_progress_nums');
        }
        $obj = new Queue;
        MqComponent::receiveQueueData('sub_group_EMP',$obj,'dealGroupRabbit');
        $procedureProgressNums = $this->redis->get('procedure_progress_nums');
        if($procedureProgressNums > 0) {
            $this->redis->decr('procedure_progress_nums');
        }
    }
    
    /**
     * 生产队列
     */
    public function actionProcedure()
    {
        set_time_limit(150);
        header("Content-type:text/html;charset=utf-8");
        
        $this->writeProcedureLog("开始进入生产队列");
        $groupId = isset($_GET['group_id']) ? intval($_GET['group_id']) : 0;
        echo "<title>生产队列</title>";
        var_dump(time() . '---111---==' . $groupId);
        $groupArr = $this->groupModel->getPendingGroup($groupId);           //   获取待处理分组
        var_dump(time() . '---222---');       
        echo '<pre>';
        $i = 1;        
        foreach ($groupArr as $group) {
            var_dump(time() . '---100s'.$i .'---');
            $this->dealGroup($group);
            $i++;
        }
        var_dump(time() . '---2002---');
        
        $this->writeProcedureLog("添加生产队列完成");
        exit("添加生产队列完成");
    }
 /**
     * 生产队列
     */
    public function actionTestProcedure()
    {
        set_time_limit(150);
        header("Content-type:text/html;charset=utf-8");
        
        $this->writeProcedureLog("开始进入生产队列");
        $groupId = isset($_GET['group_id']) ? intval($_GET['group_id']) : 0;
        echo "<title>生产队列</title>";
        var_dump(time() . '---111---==' . $groupId);
        $groupArr = $this->groupModel->getPendingGroup1($groupId);           //   获取待处理分组
        var_dump(time() . '---222---');       
        echo '<pre>';
        $i = 1;        
        foreach ($groupArr as $group) {
            var_dump(time() . '---100s'.$i .'---');
            $this->dealGroup($group);
            $i++;
        }
        var_dump(time() . '---2002---');
        
        $this->writeProcedureLog("添加生产队列完成");
        exit("添加生产队列完成");
    }
    /**
     * 处理分组
     * @param $group
     * @return void
     */
    private function dealGroup($group)
    {
        $stateArr = $this->subgroupModel->getStateCount($group['group_id']);
        $startCount = isset($stateArr[MailSubgroup::STATE_START]) ? $stateArr[MailSubgroup::STATE_START] : 0;
        $successCount = isset($stateArr[MailSubgroup::STATE_SUCCESS]) ? $stateArr[MailSubgroup::STATE_SUCCESS] : 0;
        $failCount = isset($stateArr[MailSubgroup::STATE_FAIL]) ? $stateArr[MailSubgroup::STATE_FAIL] : 0;
        $suspendCount = isset($stateArr[MailSubgroup::STATE_STOP]) ? $stateArr[MailSubgroup::STATE_STOP] : 0;
        $retryCount = isset($stateArr[MailSubgroup::STATE_START_RETRY]) ? $stateArr[MailSubgroup::STATE_START_RETRY] : 0;
//echo $group['group_id'] .'  ' .$group['total_count'] . ' '  . $startCount . ' '. $successCount. ' ' . $failCount .' '. $suspendCount.' ' . $retryCount . ' ' . '<br/>';return;
        // 如果已添加完此分组，则不用再对其进行处理
        if (0 == $suspendCount      // 检测进程清理redis里的子分组，才会在数据库里标记子分组为暂停状态。如果有标记为暂停状态的子分组，那么就会将其加入到redis中
            && 0 == $retryCount     // 即使全部发完，也可以在后台页面进行重发
            && $group['total_count'] <= ($startCount + $successCount + $failCount)
        ) {
            //echo $group['group_id'];
            return ;
        }

        // 通过group添加group_detail到redis中
        $this->addGroupDetail($group);

        // // 先发送测试用户
        // if (0 == $group['have_send_test']) {
            // $this->dealTestSendList($group);
        // }

        // 处理重发的队列
        $this->dealResendSendList($group);

        if (MailGroup::STATE_START != $group['state']) {        // 不是在启动状态，不处理新子分组和暂停子分组
            return ;
        }

        // 处理初次开始发送的队列
        $this->dealFirstTimeSendList($group);

        // 处理解除暂停的队列
        $this->dealSuspendSendList($group);
    }

    /**
     * 通过group将group_detail添加到redis中
     * @param $group
     */
    private function addGroupDetail($group)
    {
        $redis = $this->redis;
        $templateModel = new MailTemplate();

        // 获取模板
        if($group['is_ab_test'] == 1) {
            $template = $templateModel->getAbTestTemplateDetailById($group['site_id'], $group['template_id']);
            $template = array_shift($template);
        } else {
            $template = $templateModel->getTemplateById($group['template_id']);
        }
        if (empty($template)) {
            return ;
        }
        if ($group['template_pk']  != $template['id']) {    // template_pk不是最新的模板id，并修改数据库对应的记录
            $group['template_pk'] = $template['id'];
            $this->groupModel->updateGroup($group['group_id'], ["template_pk" => $template['id']]);
        }
        $group['template_subject'] = $template['template_subject'];
        $group['template_content'] = $template['template_content'];
        $group['content_text'] = '';
        $groupDetailKey = "group_detail_" . $group['group_id'];
        $groupSet = [];
        foreach ($group as $key => $value) {
            $groupSet[] = $key;
            $groupSet[] = $value;
        }
        $redis->hmset($groupDetailKey, $groupSet);
    }

    /**
     * 生产测试用户的发送列表
     * @param $group
     */
    private function dealTestSendList($group)
    {
        $info = "开始处理测试用户的发送列表<br/>";
        echo $info;
        $this->writeProcedureLog($info);

        $this->procedureLimitCheck();

        // 将分组更新为已发送测试
        $this->groupModel->updateGroup($group['group_id'], ['have_send_test' => 1]);

        $sendList = $this->sendListModel->getTestSendList($group['group_id'], $group['is_auto_group'], $group['site_id']);
        echo "测试用户的发送列表数为: " . count($sendList) . "<br/>";
        $this->writeProcedureLog("测试用户的发送列表数为: " . count($sendList));

        if (!empty($sendList)) {
            $this->makeSubgroup($group, 0, $sendList, false);
        }
    }
    
    /**
     * 检查生产进程资源的限制新版
     */
    private function procedureLimitCheckNew()
    {
        $memory = $this->redis->getMemory('used_memory') * 1;
        if ($memory > $this->config['maxRedisMemory']) {
            $msg = "超出redis最大内存使用量，退出生产进程";
            return ['code'=>1,'info'=>$msg];
        }

        if (time() - $this->startTime > $this->config['maxProcedureProcessTime']) {
            $end = \yii::$app->bjTime->date("Y-m-d H:i:s");
            $start = \yii::$app->bjTime->date("Y-m-d H:i:s", $this->startTime);
            $diff = time() - $this->startTime;

            $msg = "超过生产进程的最大运行时间: startTime[{$start}], endTime[{$end}], diff[{$diff}]s";
            return ['code'=>2,'info'=>$msg];
        }
        return ['code'=>0,'info'=>'ok'];
    }
    
    /**
     * 检查生产进程资源的限制(只检查内存)
     */
    private function procedureLimitCheckMemory()
    {
        $memory = $this->redis->getMemory('used_memory') * 1;
        if ($memory > $this->config['maxRedisMemory']) {
            $msg = "超出redis最大内存使用量，退出生产进程";
            return ['code'=>1,'info'=>$msg];
        }
        return ['code'=>0,'info'=>'ok'];
    }
    
    /**
     * 检查生产进程资源的限制(只检查时间)
     */
    private function procedureLimitCheckTime()
    {
        if (time() - $this->startTime > $this->config['maxProcedureProcessTime']) {
            $end = \yii::$app->bjTime->date("Y-m-d H:i:s");
            $start = \yii::$app->bjTime->date("Y-m-d H:i:s", $this->startTime);
            $diff = time() - $this->startTime;

            $msg = "超过生产进程的最大运行时间: startTime[{$start}], endTime[{$end}], diff[{$diff}]s";
            return ['code'=>2,'info'=>$msg];
        } 
        return ['code'=>0,'info'=>'ok'];
    }
    
    /**
     * 检查生产进程资源的限制
     */
    private function procedureLimitCheck()
    {
        $memory = $this->redis->getMemory('used_memory') * 1;
        if ($memory > $this->config['maxRedisMemory']) {
            $msg = "超出redis最大内存使用量，退出生产进程";
            $this->writeProcedureLog($msg);
            exit($msg);
        }

        if (time() - $this->startTime > $this->config['maxProcedureProcessTime']) {
            $end = \yii::$app->bjTime->date("Y-m-d H:i:s");
            $start = \yii::$app->bjTime->date("Y-m-d H:i:s", $this->startTime);
            $diff = time() - $this->startTime;

            $msg = "超过生产进程的最大运行时间: startTime[{$start}], endTime[{$end}], diff[{$diff}]s";
            $this->writeProcedureLog($msg);
            exit($msg);
        }
    }

    /**
     * 生成子分组
     * @param $group
     * @param $subgroupId
     * @param $sendList
     * @param $isRetry
     * @return bool
     */
    private function makeSubgroup($group, $subgroupId, $sendList, $isRetry)
    {
        $isNew = false;
        if (0 == $subgroupId) {
            $isNew = true;
        }
        if ($isNew) {   // 新子分组才需要生成子分组id
            $maxSubgroupId = $this->sendListModel->getMaxSubgroupId($group);
            if (false === $maxSubgroupId) {
                return false;
            }
            $subgroupId = $maxSubgroupId + 1;
        }

        $curTime = time();
        $listCount = count($sendList);
        $state = 1;     // 加入队列中

        // 更新实际发送时间（用于页面展示）
        $isUpdate = $this->groupModel->updateGroupByCondition(["group_id" => $group['group_id'], "real_send_time" => 0],
            ["real_send_time" => $curTime]);
        if ($isUpdate >= 1) {
            $this->groupModel->updateGroupByCondition(["group_id" => $group['group_id']], ["start_time" => $curTime]);
        }

        $trans = Yii::$app->db->beginTransaction();
        try {
            if ($isNew) {
                // 新子分组才需要添加子分组记录
                $this->subgroupModel->add([
                    'group_id' => $group['group_id'],
                    'subgroup_id' => $subgroupId,
                    'send_count' => $listCount,
                    'state' => MailSubgroup::STATE_START
                ]);
            } else {
                /* 旧子分组更新状态
                 * 其中有1个业务经过这里：当分组暂停时，检测进程把符合业务规则的redis分组队列清除的时候设置mysql中的子分组的状态伪暂停，
                 * 当再次启动时会加入到redis
                 */
                $this->subgroupModel->updateSubgroup($group['group_id'], $subgroupId, ['state' => MailSubgroup::STATE_START]);
            }

            // 更新sendlist记录
            $this->sendListModel->updateSubgroupAndState($group, $sendList, $subgroupId, $state);
            $trans->commit();
            $msg  = "生成子分组成功: group_id[{$group['group_id']}], subgroup_id[{$subgroupId}], count[{$listCount}]";
            $this->writeProcedureLog($msg);
            echo "$msg<br/>";
        } catch(\Exception $e) {
            $trans->rollBack();
            $msg = "生成子分组失败: group_id[{$group['group_id']}], subgroup_id[{$subgroupId}], count[{$listCount}], error: ". $e->getMessage();
            $this->writeProcedureLog($msg);
            echo "$msg<br/>";
           
            return false;
        }

        // 如果是重发,先判断子分组是否已在重发队列。如果已在重发队列中，那么就不把它加入到重发队列
        if ($isRetry) {
            $retryGroupKey = "retry_group_" . $group['group_id'];
            $retrySubgroupKey = "retry_subgroup_{$group['group_id']}_{$subgroupId}";
            $subgroupKeyArr = $this->redis->lrange($retryGroupKey, 0, -1);
            foreach ($subgroupKeyArr as $value) {
                if ($value == $retrySubgroupKey) {
                    return true;
                }
            }
        }

        $this->addSendListToRedis($group, $subgroupId, $sendList, $isRetry);
        $msg = "将子分组添加到redis: group_id[{$group['group_id']}], subgroup_id[{$subgroupId}]";
        $this->writeProcedureLog($msg);
        echo "$msg<br/>";
        return true;
    }

    /**
     * 添加发送列表
     * @param $group
     * @param $subgroupId
     * @param $sendList
     * @param $isRetry
     */
    private function addSendListToRedis($group, $subgroupId, $sendList, $isRetry = false)
    {
        $redis = $this->redis;

        // 添加子分组记录到redis
        if ($isRetry) {     // 重发
            $groupKey = "retry_group_" . $group['group_id'];
            $subgroupKey = "retry_subgroup_" . $group['group_id'] . "_" . $subgroupId;
            $mainKey = "retry_main";

            // 添加group_detail
            $detail = [];
            foreach($group as $key => $value) {
                $detail[] = $key;
                $detail[] = $value;
            }
            // $redis->hmset("retry_group_detail_" . $group['group_id'], $detail);
        } else {    // 初次发送
            $groupKey = "group_" . $group['group_id'];
            $subgroupKey = "subgroup_" . $group['group_id'] . "_" . $subgroupId;
            $mainKey = "main";
        }

        $redis->rpush($groupKey, $subgroupKey);     //  添加子分组到分组列表

        $sendListJson = [];
        foreach ($sendList as $value) {
            $value['subgroup_id'] = $subgroupId;
            $sendListJson[] = json_encode($value);
        }

        $redis->sadd($subgroupKey, $sendListJson);      // 添加sendlist
        $redis->zadd($mainKey, [$group['priority'], $groupKey]);              // 添加group到main中
    }

    /**
     * 生产初次发送的用户列表
     * @param $group
     */
    private function dealFirstTimeSendList($group)
    {
        $msg = "开始处理初次发送的用户列表: group_id[{$group['group_id']}]";
        $this->writeProcedureLog($msg);
        echo "$msg<br/>";

        while (1) {
            $this->procedureLimitCheck();

            $limit = 1000;  // 默认每次发1000条记录
            if ($group['send_per_minute']) {
                $limit = $group['send_per_minute'];
            }
            if (6 == $group['channel_id']) {    // emarsys_new通道每次1000条记录
                $limit = 1000;
                $groupSize = $this->redis->hget('emarsys_new_simulate', 'group_size');
                if (!empty($groupSize)) {
                    $limit = $groupSize * 1;
                }
            }
            $sendList = $this->sendListModel->getFirstTimeSendList($group, $group['is_auto_group'], $limit);
            if (empty($sendList)) {
                return;
            }

            if (!empty($sendList)) {
                $this->makeSubgroup($group, 0, $sendList, false);
            }
        }
    }

    /**
     * 生产解除暂停的队列
     * @param $group
     */
    private function dealSuspendSendList($group)
    {
        $msg = "开始处理解除暂停的队列: group_id[{$group['group_id']}]";
        $this->writeProcedureLog($msg);
        echo "$msg<br/>";

        $subgroupArr = $this->subgroupModel->getSuspendSubgroup($group['group_id']);
        if (empty($subgroupArr)) {
            return ;
        }

        foreach ($subgroupArr as $subgroup) {
            $this->procedureLimitCheck();
            $sendList = $this->sendListModel->getSendList($group['site_id'], $group['group_id'],
                $subgroup['subgroup_id'], $group['is_auto_group']);
            if (!empty($sendList)) {
                $this->makeSubgroup($group, $subgroup['subgroup_id'], $sendList, false);
            }
        }
    }

    /**
     * 生产重发的队列
     * @param $group
     */
    public function dealResendSendList($group)
    {
        $msg = "开始处理重发的队列: group_id[{$group['group_id']}]";
        $this->writeProcedureLog($msg);
        echo "$msg<br/>";

        $subgroupArr = $this->subgroupModel->getRetrySubgroup($group['group_id']);
        if (empty($subgroupArr)) {
            return ;
        }

        foreach ($subgroupArr as $subgroup) {
            $this->procedureLimitCheck();
            $sendList = $this->sendListModel->getSendList($group['site_id'], $group['group_id'],
                $subgroup['subgroup_id'], $group['is_auto_group']);
            if (!empty($sendList)) {
                $this->makeSubgroup($group, $subgroup['subgroup_id'], $sendList, true);
            }
        }
    }

    /**
     * 消费master
     */
    public function actionCustomerMaster()
    {
        ignore_user_abort(true);     // 忽略用户的中断
        set_time_limit(300);

        // fixme: machineNum、workerNum...都要从数据库中获取。
        header("Content-type:text/html;charset=utf-8");
        $machineNum = $this->config['machineNum'];        // 机器数量
        $workerNum = $this->config['workerNum'];         // 每台机器的worker数
        $sendSpeed = $this->config['sendSpeed'];        // 每分钟发送的子分组数
        $workerMaxAliveTime = $this->config['workerMaxAliveTime']; // worker的最大存活时间，单位：秒
        $cookieArr = $this->config['cookieArr'];            // 用于负载均衡的cookie设置

        echo "<title>消费master</title>";

        // 从redis中获取父分组
        $redis = $this->redis;
        $availGroupArr = $this->getAvailGroup("main");

        if (empty($availGroupArr)) {
            exit("没有需要处理的分组");
        }

        $lock = new FileLock("master");
        if (false === $lock->tryLock()) {
            exit("获取文件锁失败，master进程已在运行");
        }

        // 获取各子分组数量
        $totalSubgroupCount = 0;
        foreach ($availGroupArr as $key => $value) {
            $subGroupCount = $this->redis->llen($key);
            if (0 == $subGroupCount) {      // 子分组不存在，继续处理下一个分组
                unset($availGroupArr[$key]);
                continue;
            }

            $availGroupArr[$key]['subgroupCount'] = $subGroupCount;
            $totalSubgroupCount += $subGroupCount;
        }

        if (0 == $totalSubgroupCount) {
            exit("没有需要处理的子分组");
        }

        // 计算worker数组
        $workerArr = $this->calcWorkArr($machineNum, $workerNum, $workerMaxAliveTime);
      
        // 把worker分配给分组
        $this->allocateWorker($sendSpeed, $availGroupArr, $workerArr);
      
//        var_dump($workerArr);exit;
        // 依次启动worker
        foreach ($workerArr as $worker) {
            if (!isset($worker['group'])) {
                continue;
            }
            $url = "http://{$_SERVER['HTTP_HOST']}/marketing/queue/customer-worker"
                . "?module_name=marketing_email&process={$worker['processName']}&group={$worker['group']}";

            if (isset($_GET['simulate']) && !empty($_GET['simulate'])) {
                $url = $url . "&simulate=1";
            }
            $cookie = $cookieArr[$worker['machineId']];
            $item['url'] = $url;
            $item['options'][CURLOPT_COOKIE] = $cookie;
            $item['options'][CURLOPT_NOSIGNAL] = 1;         // 不使用信号，让worker可以小于1s启动
            $item['options'][CURLOPT_TIMEOUT_MS] = 100;      // 给足够的时间worker启动
            yii::$app->curl->execute($item);
            echo "添加worker: {$worker['processName']}, url[{$url}]<br/>";
        }

        echo "master分配完worker";
    }

    /**
     * 获取需要处理的父分组
     * @param $redisKey         key(main或者retry_main)
     * @return array
     */
    private function getAvailGroup($redisKey)
    {
       
        $zset = $this->redis->zrange($redisKey, 0, -1, true);
        
       
        // group的key为组名(如group_1)
        $group = [];
        foreach ($zset as $key => $value) {
            if (0 == $key % 2) {
                if ($this->config['stopPriority'] == $zset[$key+1]) {   // 分组为暂停分组，不用处理
                    continue;
                }
                
                
                $arr = explode("_", $value);
                $groupId = $arr[count($arr)-1];
                $group[$value] = ["score" => $zset[$key+1], "group_id" => $groupId];
            }
        }

        return $group;
    }

    /**
     * 计算worker数组。worker数组中存储的是可用于分配的worker资源
     * @param $machineNum               机器数量
     * @param $workerNum                每台机器的worker数
     * @param $workerMaxAliveTime       worker的最大存活时间
     * @return array
     */
    private function calcWorkArr($machineNum, $workerNum, $workerMaxAliveTime)
    {
        $freeWorkerArr = [];    // 空闲worker
        $notFreeWorkArr = [];   // 非空闲worker
        $redis = $this->redis;
        $redisArr = $redis->hgetall("process");
        $commonArr = $redis->redisArrToCommonArr($redisArr);
        $cur = time();

        for ($i = 1; $i <= $workerNum; $i++) {
            for ($j = 1; $j <= $machineNum; $j++) {
                $processName = "process_{$j}_{$i}";
                $processUseTime = 0;        // 进程的最后使用时间
                if (isset($commonArr[$processName])) {
                    $data = json_decode($commonArr[$processName], true);
                    $processUseTime = $data['use_time'];
                }

                // 如果进程名不存在，进程已退出，进程超时。我们就将这些进程加入到worker数组
                if ( 0 == $processUseTime
                    || $cur - $processUseTime > $workerMaxAliveTime  // $workerMaxAliveTime 是在worker的setlimittime
                ) {
                    $freeWorkerArr[] = ['processName' => $processName, 'machineId' => $j];
                } else {
                    $notFreeWorkArr[] = ['processName' => $processName, 'machineId' => $j];
                }

            }
        }
        $workerArr = array_merge($freeWorkerArr, $notFreeWorkArr);
        return $workerArr;
    }

    /**
     * 分配worker
     * @param $sendSpeed
     * @param $availGroup
     * @param $workerArr
     */
    private function allocateWorker($sendSpeed, $availGroup, &$workerArr)
    {
        $workerIndex = 0;
        $workerCount = count($workerArr);
        $redis = $this->redis;

        foreach ($availGroup as $key => $value) {
            if ($workerIndex >= $workerCount) {
                return ;
            }

            $groupDetailKey = "group_detail_" . $value['group_id'];
            $detail = $redis->hgetall($groupDetailKey);
            $groupDetail = $redis->redisArrToCommonArr($detail);
            if (empty($groupDetail)) {      // 异常的预防
                continue;
            }
            
           
           if (!empty($groupDetail['error_code']) && in_array($groupDetail['error_code'],$this->config['errorCode']) && !empty($groupDetail['group_last_send']) && time() - $groupDetail['group_last_send'] <  $this->config['sendSpacingTime']) {
              
                $this->writeLog($groupId, "[{$groupDetail['group_id']}]上一次发送错误码为{$groupDetail['error_code']}, 并且未达到重发的间隔时间,上次重发/发送时间为" . date("Y-m-d H:i:s",$groupDetail['group_last_send']));
                echo "超限：".$groupDetail['group_id']."<br>";
                continue;
            }

            // 指定了该分组分配的worker数，按指定worker数分配worker
            if (isset($groupDetail['worker_num']) && $groupDetail['worker_num'] > 0) {
                for ($i = 0; $i < $groupDetail['worker_num']; $i++) {
                    if ($workerIndex < $workerCount) {
                        $workerArr[$workerIndex]['group'] = "group_" . $groupDetail['group_id'];
                        ++$workerIndex;
                    } else {
                        break;
                    }
                }
                continue;
            }

            // 按指定时间发完分组来分配worker
            $allocateNum = 1;
            $len = $this->redis->llen("group_" . $groupDetail['group_id']);
            if (0 == $groupDetail['dead_line']) {       // 没有指定截止发送时间，那么给该分组分配一个worker
                $allocateNum = 1;
            } else {
                $remainTime = floor(($groupDetail['dead_line'] - time()) / 60);  // 剩余时间（单位：分钟）
                if ($remainTime <= 0) {
                    $allocateNum = ceil($len / $sendSpeed);
                } else {
                    $allocateNum = ceil(($len / $remainTime) / $sendSpeed);
                }
            }

            for ($i = 0; $i < $allocateNum; $i++) {
                if ($workerIndex < $workerCount) {
                    $workerArr[$workerIndex]['group'] = "group_" . $groupDetail['group_id'];
                    ++$workerIndex;
                } else {
                    break;
                }
            }

        }
    }

    /**
     * 消费worker
     */
    public function actionCustomerWorker()
    {
        
        // fixme: 此部分参数从外部传入
        header("Content-type:text/html;charset=utf-8");
        $maxRetryTime = $this->config['maxGroupFailCount'];     // 最大重发次数
        $maxExecuteTime = $this->config['maxWorkerProcessTime'];   // 最大执行时间（单位：秒）

        ignore_user_abort(true);     // 忽略用户的中断  包括curl
        // 最多执行5分钟  当一个进程卡主(mysql锁等待)、连外部资源（redis、msyql）的时候超时,需要给足时间来响应，
        //同时也限制新的进程产生，虽然影响了发送效率但是小概率事件，如果是连外部资源（redis、msyql）的时候超时，新进程也无法运作
        set_time_limit(300);

        echo "<title>消费worker</title>";

        // 验证参数合法性
        $processName = yii::$app->request->get("process");
        $groupKey = yii::$app->request->get("group");
        if (empty($processName) || empty($groupKey)) {
            exit("参数不完整！");
        }
        $arr = explode("_", $groupKey);
        if (2 != count($arr) && !is_numeric($arr[1])) {
            exit("group参数格式不正确");
        }

        $lock = new FileLock($processName);
        if (false === $lock->tryLock()) {
            exit("获取文件锁失败，已有相同worker[{$processName}]进程启动");
        }

        $groupId = $arr[1];
        $groupDetailKey = "group_detail_" . $groupId;

        $redis = $this->redis;

        // 更新worker进程标识的时间
        $startTime = time();
        $processData = ['start_time' => $startTime, 'use_time' => $startTime, 'group_id' => $groupId];
        $processJsonData = json_encode($processData);
        $redis->hset("process", $processName, $processJsonData);
        $sendSubgroupNum = 0;   // 记录当前worker发送子分组数，用于写日志
        
        while (1) {
            if (0 == $redis->exists($groupDetailKey)) {
                $redis->zrem("main", [$groupKey]);
                break;
            }

            $detail = $redis->hgetall($groupDetailKey);
            $groupDetail = $redis->redisArrToCommonArr($detail);
            
            if (!empty($groupDetail['error_code']) && in_array($groupDetail['error_code'],$this->config['errorCode']) && !empty($groupDetail['group_last_send']) && time() - $groupDetail['group_last_send'] <  $this->config['sendSpacingTime']) {
              
                $this->writeLog($groupId, "[{$groupDetail['group_id']}]上一次发送错误码为{$groupDetail['error_code']}, 并且未达到重发的间隔时间,上次重发/发送时间为" . date("Y-m-d H:i:s",$groupDetail['group_last_send']));
                echo "超限：".$groupDetail['group_id']."<br>";
                exit;
            }


            if (isset($groupDetail['fail_count']) && $groupDetail['fail_count'] >= $maxRetryTime) {
                exit("重发数大于{$maxRetryTime}次，停止发送分组:{$groupKey}");
            }

            if ($groupDetail['priority'] == $this->config['stopPriority']) {
                exit("分组{$groupKey}的状态为暂停发送，暂停发送此分组");
            }

            // 获取子分组
            $subgroupKey = $redis->lpop($groupKey);
            //增加在数据库判断子分组是否已发送成功,如果成功就直接跳过
            if($subgroupKey !== null){
                $arr = explode("_", $subgroupKey);
                $subgroupId = $arr[count($arr)-1];
                if($this->subgroupModel->getSuccessSubgroup($groupDetail['group_id'],$subgroupId)){
                    $this->writeLog($groupDetail['group_id'], "数据表子分组[{$subgroupId}]已发送");
                    continue;
                }
            }
            
            $this->writeLog($groupId, "获取子分组[{$subgroupKey}]");

            try {
                if (null === $subgroupKey) {
                    $isRem = $redis->zrem("main", $groupKey);
                    $this->writeLog($groupId, "删除分组[{$groupId}], isRem[{$isRem}]");
                    break;
                }

                $this->sendSubgroup($processName, $groupDetail, $subgroupKey);
                sleep(1);       // 休眠1秒，防止进程挂掉
                $sendSubgroupNum++;
            } catch (\Exception $e) {
                $this->writeLog($groupId, "worker: 发送分组[{$groupId}]异常[" . $e->getMessage() . "]");

                // 如果还存在子分组的sendList，那么就说明没有发送成功。此时需要把subgroupKey补充回groupKey列表
                if ($redis->exists($subgroupKey)) {
                    $redis->rpush($groupKey, $subgroupKey);
                }
            }


            $curTime = time();
            $processData = ['start_time' => $startTime, 'use_time' => $curTime, 'group_id' => $groupId];
            $processJsonData = json_encode($processData);
            $redis->hset("process", $processName, $processJsonData);

            if ($curTime - $startTime > $maxExecuteTime) {
                break;
            }
        }

        // 更新当前进程标识，将其设置为退出进程状态
        $processData = ['start_time' => 0, 'use_time' => 0, 'group_id' => ""];
        $processJsonData = json_encode($processData);
        $redis->hset("process", $processName, $processJsonData);

        $msg = "exit worker. process: {$processName}, group: {$groupKey}, 发送子分组数[{$sendSubgroupNum}]";
        $this->writeLog($groupId, $msg);
    }

    /**
     * 生成模板
     * @param $groupDetail
     * @return mixed
     */
    private function makeTemplate($groupDetail)
    {
        $mailSubject = $groupDetail['template_subject'];  // 邮件标题
        $mailContent = $groupDetail['template_content'];

        // 替换特殊符号
        $mailContent = str_replace("&quot;", "\"", $mailContent);
        $mailContent = str_replace("&lt;", "<", $mailContent);
        $mailContent = str_replace("&gt;", ">", $mailContent);
        $mailContent = str_replace("&raquo;", ">", $mailContent);
        $mailContent = str_replace("&amp;", "&", $mailContent);
        $mailContent = str_replace('{track_code}', $groupDetail['track_code'], $mailContent);
        $mailContent = str_replace('{$mail_utm_source}','mail_api', $mailContent);  // GA 发送通道设为emarsys
        
        // 如果有other_vars则也要替换
        if (!empty($groupDetail['other_vars'])) {
            $otherVars = json_decode($groupDetail['other_vars'],true);
            if (!empty($otherVars)) {   // 替换模板内容和模板标题中的变量
                if (!empty($otherVars['email_title'])) {
                    $mailSubject = str_replace('{email_title}',$otherVars['email_title'] , $mailSubject);
                }
                
                for ($i = 1;$i <= 12; $i++) {
                    if (!empty($otherVars['goods_url'.$i])) {
                        $mailContent = str_replace("{goods_url{$i}}", $otherVars['goods_url'.$i], $mailContent);
                    } else {
                        $mailContent = str_replace("{goods_url{$i}}", '', $mailContent);
                    }
                    if (!empty($otherVars['goods_img'.$i])) {
                        $mailContent = str_replace("{goods_img{$i}}", $otherVars['goods_img'.$i], $mailContent);
                    } else {
                        $mailContent = str_replace("{goods_img{$i}}", '', $mailContent);
                    }
                    if (!empty($otherVars['goods_price'.$i])) {
                        $mailContent = str_replace("{goods_price{$i}}", $otherVars['goods_price'.$i], $mailContent);
                    } else {
                        $mailContent = str_replace("{goods_price{$i}}", '', $mailContent);
                    }
                    if (!empty($otherVars['goods_title'.$i])) {
                        $mailContent = str_replace("{goods_title{$i}}", $otherVars['goods_title'.$i], $mailContent);
                    } else {
                        $mailContent = str_replace("{goods_title{$i}}", '', $mailContent);
                    }
                }
            }
        }

        if (2 == $groupDetail['channel_id'] || 6 == $groupDetail['channel_id']) {   // 替换emarsys、emarsys_new通用的模板变量
            $mailSubject = preg_replace('/\{\$(.*?)\}/s', '$$\\1$$', $mailSubject);     // 替换subject变量
            $mailContent = str_replace('{$name}','$$firstname$$', $mailContent);  // 邮件内容
            $mailContent = str_replace('{$site_id}','$$site_id$$', $mailContent);  // 邮件内容
            $mailContent = str_replace('{$email}','$$email$$', $mailContent);  // 邮件内容
            $mailContent = str_replace('{$site_user_id}','$$CUSTOMERid$$', $mailContent);  // 网站端的user_id
            $mailContent = str_replace('{$email_encode}','$$email_encode$$', $mailContent);  // 邮件内容
            $mailContent = str_replace('$gid',$groupDetail['group_id'], $mailContent);  // 邮件内容
            $mailContent = preg_replace('/\{\$(.*?)\}/s', '$$\\1$$', $mailContent);     // 替换其他变量
            
            if($groupDetail['channel_id'] == 2) {
                // 添加打开邮件代码
                $mailContent .= '<div><img src=\'http://ems.appinthestore.com/marketing/mail-user-deal/open?'
                        . 'module_name=marketing_email&m=$$mail_id2$$&u=$$user_id$$&u1=$$CUSTOMERid$$&t=$$email_type$$&g='
                        .$groupDetail['group_id']."&s=".$groupDetail['site_id'].'&token=$$token$$'."&act=read' height='0' width='0'/></div>";
            } else if($groupDetail['channel_id'] == 6) {
                // 添加打开邮件代码
                $mailContent .= '<div><img src=\'http://ems.appinthestore.com/marketing/mail-user-deal/open?'
                        . 'module_name=marketing_email&m=$$mail_id2$$&u=$$user_id$$&u1=$$site_user_id$$&t=$$email_type$$&g='
                        .$groupDetail['group_id']."&s=".$groupDetail['site_id'].'&token=$$token$$'."&act=read' height='0' width='0'/></div>";
            }
            
        } else if (4 == $groupDetail['channel_id']) {   // cheet通道
            $mailSubject = preg_replace('/\{\$(.*?)\}/s', '%%\\1%%', $mailSubject);     // 替换subject变量
            $mailContent = str_replace('{$site_id}','%%TAG_2%%', $mailContent);  //邮件内容
            $mailContent = str_replace('{$email}','%%EMAIL%%', $mailContent);  //邮件内容
            $mailContent = str_replace('{$email_encode}','&email=%%TAG_3%%', $mailContent);  //邮件内容
            $mailContent = str_replace('$gid', $groupDetail['group_id'], $mailContent);  //邮件内容
            $mailContent = preg_replace('/\{\$(.*?)\}/s', '%%\\1%%', $mailContent);     // 替换其他变量

            // 添加打开邮件代码
            $mailContent .= '<div><img src=\'http://ems.appinthestore.com/marketing/mail-user-deal/open?'
                . 'module_name=marketing_email&m=%%TAG_1%%&g='
                .$groupDetail['group_id'] ."&s=".$groupDetail['site_id'].'&token=%%token%%'."&act=read' height='0' width='0'/></div>";
        } else if (5 == $groupDetail['channel_id']) { // benchmark通道
            // 替换emarsys、emarsys_new通用的模板变量
            $mailSubject = str_replace('{$firstname}','[contact_attribute:firstname]', $mailSubject); 
            $mailSubject = preg_replace('/\{\$(.*?)\}/s', '[contact_attribute:\\1]', $mailSubject);     // 替换subject变量
            
            $mailContent = str_replace('{$name}','[contact_attribute:firstname]', $mailContent);  // 邮件内容
            $mailContent = str_replace('{$firstname}','[contact_attribute:firstname]', $mailContent);  // 邮件内容
            // $mailContent = str_replace('{$site_id}','[contact_attribute:extra 6]', $mailContent);  // 邮件内容
            $mailContent = str_replace('{$email}','[contact_attribute:email]', $mailContent);  // 邮件内容
            // $mailContent = str_replace('{$site_user_id}','[contact_attribute:extra 4]', $mailContent);  // 网站端的user_id
            // $mailContent = str_replace('{$email_encode}','[contact_attribute:extra 6]', $mailContent);  // 邮件内容
            $mailContent = str_replace('{$code}','[contact_attribute:extra 6]', $mailContent);  // 邮件内容
            $mailContent = str_replace('$gid',$groupDetail['group_id'], $mailContent);  // 邮件内容
            $mailContent = preg_replace('/\{\$(.*?)\}/s', '[contact_attribute:\\1]', $mailContent);     // 替换其他变量

            // 添加打开邮件代码
            $mailContent .= '<div><img src=\'http://ems.appinthestore.com/marketing/mail-user-deal/open?'
                    . 'module_name=marketing_email&m=[contact_attribute:extra 3]&u=[contact_attribute:extra 4]&u1=[contact_attribute:Phone]&t=[contact_attribute:State]&g='
                    .$groupDetail['group_id']."&s=".$groupDetail['site_id'].'&token=[contact_attribute:extra 5]'."&act=read' height='0' width='0'/></div>";
            
        } else if (9 == $groupDetail['channel_id']) { // webpower通道

            // 添加打开邮件代码
            $openCode .= '<div><img src=\'http://ems.appinthestore.com/marketing/mail-user-deal/open?'
                    . 'module_name=marketing_email&m={$mail_id2}&u={$user_id}&u1={$CUSTOMERid}&t={$email_type}&g='
                    .$groupDetail['group_id']."&s=".$groupDetail['site_id'].'&token={$token}'."&act=read' height='0' width='0'/></div>";
            
            $mailContent = str_replace('</body>',$openCode . '</body>', $mailContent);
            
        }

        $template['subject'] = $mailSubject;
        $template['content'] = $mailContent;

        return $template;
    }

    /**
     * 发送子分组
     * @param $groupDetail
     * @param $processName
     * @param $subgroupKey
     */
    private function sendSubgroup($processName, $groupDetail, $subgroupKey)
    {
        $redis = $this->redis;
        $members = $redis->smembers($subgroupKey);
        if (empty($members)) {
            $this->writeLog($groupDetail['group_id'], "发送子分组[{$subgroupKey}]，没有sendList");
            return ;
        }

        $sendList = [];
        $channel = new Channel();
        $groupModel = $this->groupModel;
        $arr = explode("_", $subgroupKey);
        $subgroupId = $arr[count($arr)-1];
        $template = $this->makeTemplate($groupDetail);
        $info['template'] = $template;
        $info['groupDetail'] = $groupDetail;
        $info['subgroupId'] = $subgroupId;
        $info['processId'] = str_replace("process_", "", $processName);

        $this->writeLog($groupDetail['group_id'], "子分组[{$subgroupId}]，开始json_decode解码发送列表");
        foreach ($members as $value) {
            $sendList[] = json_decode($value, true);
        }

        $this->writeLog($groupDetail['group_id'], "开始发送子分组[{$subgroupId}]");
        $result = $channel->send($groupDetail['channel_id'], $info, $sendList);        // 发送列表

        if (true == $result) {  // 发送成功
            // 更新redis和数据库。如果第一次失败，那么再重试两次。
            $this->writeLog($groupDetail['group_id'], "发送子分组[{$subgroupId}]成功");
            for ($i = 1; $i <= 3; ++$i) {
                $isUpdate = $this->updateSubgroupSendSuccess($groupDetail, $subgroupId, $sendList);
                $this->writeLog($groupDetail['group_id'], "更新发送子分组[{$subgroupId}]第{$i}次，isUpdate[{$isUpdate}]。");
                if ($isUpdate) {
                    break;
                }
                sleep($i);
            }

        } else {
            $this->writeLog($groupDetail['group_id'], "发送子分组[{$subgroupId}]失败");
            $this->updateSubgroupFail($groupDetail, $subgroupId, 1);

            // 加入重发队列
            $isRename = $redis->rename($subgroupKey, "retry_" . $subgroupKey);
            $isAdd = $redis->zadd("retry_main", [$groupDetail['priority'], "retry_group_" . $groupDetail['group_id']]);
            $llen = $redis->rpush("retry_group_"  . $groupDetail['group_id'], "retry_" . $subgroupKey);
            $log = "worker添加重发队列，子分组[{$subgroupId}],rename[{$isRename}],zadd[{$isAdd}],重发分组长度[{$llen}]";
            $this->writeLog($groupDetail['group_id'], $log);
        }

        $isUpdate = $groupModel->updateGroup($groupDetail['group_id'], ["group_last_send" => time()]);
        if($groupDetail['is_ab_test'] == 1 && $groupDetail['ab_test_parent_id'] > 1) {
            $sendOkCounts = $groupModel->getTestChildsSendOkCount($groupDetail['ab_test_parent_id']);
            $groupModel->updateGroup($groupDetail['ab_test_parent_id'], ["group_last_send" => time(),'send_ok_count' => $sendOkCounts]);
            if(!$groupModel->getIsAllSend($groupDetail['ab_test_parent_id'])) {
                $groupModel->updateGroup($groupDetail['ab_test_parent_id'], ["state" => 2]);
            }
        }
        $this->writeLog($groupDetail['group_id'], "更新最后发送时间是否成功[{$isUpdate}]");

    }

    /**
     * 更新发送成功子分组的相关数据
     * @param $groupDetail
     * @param $subgroupId
     * @param $sendList
     * @param bool $isRetry     是否重发
     * @return bool
     */
    private function updateSubgroupSendSuccess($groupDetail, $subgroupId, $sendList, $isRetry = false)
    {
        $redis = $this->redis;
        $groupModel = $this->groupModel;
        $subgroupKey = "subgroup_" . $groupDetail['group_id'] . "_" . $subgroupId;
        $groupKey = "group_" . $groupDetail['group_id'];
        $mainKey = "main";
        if ($isRetry) {
            $subgroupKey = "retry_" . $subgroupKey;
            $groupKey = "retry_" . $groupKey;
            $mainKey = "retry_" . $mainKey;
        }

        // 在redis中删除send list
        $isDel = $redis->del($subgroupKey);
        $groupLen = $redis->llen($groupKey);
        $this->writeLog($groupDetail['group_id'], "redis删除子分组[{$subgroupId}], isDel[{$isDel}], 分组长度[{$groupLen}]");
        if (0 == $groupLen) {
            $isRem = $redis->zrem($mainKey, [$groupKey]);
            $this->writeLog($groupDetail['group_id'], "redis删除分组[{$groupDetail['group_id']}], isRem[{$isRem}]");
        }

        // 有一种坑，开启事务回滚之后，后台暂停很久后启动，检测进程清除内存又加进队列会再次发送导致发多次，如果没有暂停，那么数据库的状态都是发送中，但实际上是发送了
        $trans = Yii::$app->db->beginTransaction();
        try {
            // 更新数据库 (现在状态更新放到subgroup表中 这个代码不用了)
//            $this->sendListModel->updateSendList($groupDetail, $subgroupId,
//                ['state' => 2, 'last_send' => time(), 'template_pk' => $groupDetail['template_pk']]);   // 更新发送成功的sendlist
            $count = count($sendList);
            $groupModel->updateOkCount($groupDetail, $count);
            $this->subgroupModel->updateSubgroup($groupDetail['group_id'], $subgroupId, ['state' => MailSubgroup::STATE_SUCCESS,'last_send' => time(), 'template_pk' => $groupDetail['template_pk']]);
            $finishCount = $this->subgroupModel->getFinishCount($groupDetail['group_id']);
            $groupModel->updateStateByCount($groupDetail['group_id'], $finishCount);
            $curTime = time();
            $isUpdate = $groupModel->updateGroupByCondition(["group_id" => $groupDetail['group_id'], "real_send_time" => 0],
            ["real_send_time" => $curTime]);
            
            if ($isUpdate >= 1) {
                $this->groupModel->updateStartTimeByGroupId($groupDetail['group_id']);
            }
            $trans->commit();
            $this->writeLog($groupDetail['group_id'], "更新数据库成功: 子分组[{$subgroupId}], count[{$count}], finishCount[$finishCount]");
        } catch (\Exception $e) {
            $trans->rollBack();
            $this->writeLog($groupDetail['group_id'], "更新数据库失败，回滚数据库：子分组[{$subgroupId}]");
            return false;
        }

        $isExistsRetryGroup = $redis->exists("retry_group_". $groupDetail['group_id']);
        // 如果已完成的数量超过总数并且没有相应的重发分组，那么删除group_detail key。其中group_detail key是worker和resend进程共用的
        if ($groupDetail['total_count'] <= $finishCount && 0 == $isExistsRetryGroup) {
            $isDel = $redis->del("group_detail_" . $groupDetail['group_id']);
            $this->writeLog($groupDetail['group_id'], "redis删除group_detail, isDel[{$isDel}]");
        }

        // 更新user表的最后发送时间
        $userModel = new User();
        $userModel->updateLastSentTime($groupDetail['site_id'], $groupDetail['group_id'], $subgroupId);
        return true;
    }

    /**
     * 在redis中对子分组失败次数加上指定的增量
     * @param $groupDetail
     * @param $subgroupId
     * @param $count
     * @return mixed
     */
    private function updateRedisSubgroupFailCount($groupDetail, $subgroupId, $count)
    {
        $redis = $this->redis;
        $groupDetailKey = "group_detail_" . $groupDetail['group_id'];
        $groupFailKey = "fail_count";
        $subgroupFailKey = "subgroup_" . $subgroupId . "_fail_count";
        $redis->hsetnx($groupDetailKey, [$groupFailKey, 0]);
        $failCountArr['group_fail_count'] = $redis->hincrby($groupDetailKey, [$groupFailKey, $count]);
        $redis->hsetnx($groupDetailKey, [$subgroupFailKey, 0]);
        $failCountArr['subgroup_fail_count'] = $redis->hincrby($groupDetailKey, [$subgroupFailKey, $count]);
        return $failCountArr;
    }
    
    /**
     * 对子分组失败次数加上指定的增量（包括redis和数据库）
     * @param $groupDetail
     * @param $subgroupId
     * @param $count
     * @return mixed
     */
    private function updateSubgroupFailCount($groupDetail, $subgroupId, $count)
    {
        $failCountArr = $this->updateRedisSubgroupFailCount($groupDetail, $subgroupId, $count);

        $isUpdate1 = $this->sendListModel->updateFailCount($groupDetail, $subgroupId, $count);
        $isUpdate2 = $this->subgroupModel->updateFailCount($groupDetail['group_id'], $subgroupId, $count);
        $log = "更新redis: 父分组失败数[{$failCountArr['group_fail_count']}]，"
            . "子分组失败数[{$failCountArr['subgroup_fail_count']}]";
        $this->writeLog($groupDetail['group_id'], $log);
        $this->writeLog($groupDetail['group_id'], "更新数据库：子分组失败次数是否更新[{$isUpdate2}]，sendList更新数[{$isUpdate1}]");
        return $failCountArr;
    }

    /**
     * 更新子分组失败相关数据
     * @param $groupDetail
     * @param $subgroupId
     * @param $count
     * @return array
     */
    private function updateSubgroupFail($groupDetail, $subgroupId, $count)
    {
        $failCountArr = $this->updateSubgroupFailCount($groupDetail, $subgroupId, $count);
        $this->subgroupModel->updateSubgroup($groupDetail['group_id'], $subgroupId, ['state' => MailSubgroup::STATE_FAIL]);
        $finishCount = $this->subgroupModel->getFinishCount($groupDetail['group_id']);
        $this->groupModel->updateStateByCount($groupDetail['group_id'], $finishCount);
        $this->writeLog($groupDetail['group_id'], "更新子分组[{$subgroupId}]失败相关数据, FinishCount[{$finishCount}]。");
        return $failCountArr;
    }

    /**
     * 获取后缀。用于获取groupKey的group_id, subgroupKey的subgroup_id
     * @param $key
     * @param string $delimiter         分隔符
     * @return mixed
     */
    private function getSuffix($key, $delimiter = '_')
    {
        $arr = explode($delimiter, $key);
        return $arr[count($arr)-1];
    }

    /**
     * 发送子分组
     * @param $groupDetail
     * @param $processName
     * @param $retrySubgroupKey
     */
    private function sendRetrySubgroup($processName, $groupDetail, $retrySubgroupKey)
    {
        // fixme: 下面的参数应该从配置中读取
        $maxFailCount = $this->config['maxSubgroupFailCount'];

        $redis = $this->redis;

        $members = $redis->smembers($retrySubgroupKey);
        $membersCount = count($members);
        $this->writeLog($groupDetail['group_id'], "重发：subgroup[{$retrySubgroupKey}], 子分组的邮件数[{$membersCount}]");
        if (empty($members)) {
            return ;
        }

        $sendList = [];
        $channel = new Channel();
        $subgroupModel = new MailSubgroup();
        $subgroupId = $this->getSuffix($retrySubgroupKey);
        $template = $this->makeTemplate($groupDetail);
        $info['template'] = $template;
        $info['groupDetail'] = $groupDetail;
        $info['subgroupId'] = $subgroupId;
        $info['processId'] = str_replace("process_", "", $processName);

        foreach ($members as $value) {
            $sendList[] = json_decode($value, true);
        }

        $this->groupModel->updateGroup($groupDetail['group_id'], ["group_last_send" => time()]);

        $this->writeLog($groupDetail['group_id'], "重发: 子分组[{$subgroupId}]开始通过渠道[{$groupDetail['channel_id']}]发送邮件");
        $result = $channel->send($groupDetail['channel_id'], $info, $sendList);        // 发送列表

        if (true == $result) {
            $this->writeLog($groupDetail['group_id'], "重发：子分组[{$subgroupId}]发送成功");

            // 在redis中减去父分组、子分组的失败次数。以便worker可以知道当前还在处理的fail_count减少
            $subgroupFailCountKey = "subgroup_{$subgroupId}_fail_count";
            $failCount = isset($groupDetail[$subgroupFailCountKey]) ?  $groupDetail[$subgroupFailCountKey]*(-1) : 0;
            $this->updateRedisSubgroupFailCount($groupDetail, $subgroupId, $failCount);

            // 更新成功发送分组的相关数据
            $isUpdate = $this->updateSubgroupSendSuccess($groupDetail, $subgroupId, $sendList, true);
            if($groupDetail['is_ab_test'] == 1 && $groupDetail['ab_test_parent_id'] > 1) {
                $sendOkCounts = $this->groupModel->getTestChildsSendOkCount($groupDetail['ab_test_parent_id']);
                $this->groupModel->updateGroup($groupDetail['ab_test_parent_id'], ["group_last_send" => time(),'send_ok_count' => $sendOkCounts]);
                
                if(!$this->groupModel->getIsAllSend($groupDetail['ab_test_parent_id'])) {
                    $this->groupModel->updateGroup($groupDetail['ab_test_parent_id'], ["state" => 2]);
                }
            }
            $this->writeLog($groupDetail['group_id'], "重发：子分组[{$subgroupId}]，更新相关成功数据 isUpdate[{$isUpdate}]");
        } else {
            $this->writeLog($groupDetail['group_id'], "重发：子分组[{$subgroupId}]发送失败");

            $failCountArr = $this->updateSubgroupFail($groupDetail, $subgroupId, 1);

            // 如果子分组失败计数超过临界值，那么记录失败同时向rms发送告警。
            if ($failCountArr['subgroup_fail_count'] >= $maxFailCount) {       // // 如果子分组失败计数超过临界值，放弃重发
                $isUpdate = $this->updateRetrySubgroupFail($groupDetail, $subgroupId, $failCountArr);
                $this->writeLog($groupDetail['group_id'], "重发：子分组[{$subgroupId}]，更新相关失败数据 isUpdate[{$isUpdate}]");

                // todo: 向rms发送告警
                $log = "发送子分组[{$retrySubgroupKey}]，失败次数[{$failCountArr['subgroup_fail_count']}]，放弃重试。<br/>";
                $this->writeLog($groupDetail['group_id'], $log);
                echo $log;
            } else {        // 否则，将子分组重新添加到重发队列。redis: subgroupKey add。
                $redis->rpush("retry_group_"  . $groupDetail['group_id'], $retrySubgroupKey);
            }
        }

    }

    /**
     * 更新重发失败子分组的数据
     * @param $groupDetail
     * @param $subgroupId
     * @param $failCountArr
     */
    private function updateRetrySubgroupFail($groupDetail, $subgroupId, $failCountArr)
    {
        $redis = $this->redis;
        $groupModel = $this->groupModel;
        $subgroupKey = "retry_subgroup_" . $groupDetail['group_id'] . "_" . $subgroupId;
        $groupKey = "retry_group_" . $groupDetail['group_id'];
        $mainKey = "retry_main";
        $groupDetailKey = "group_detail_" . $groupDetail['group_id'];
        $groupFailKey = "fail_count";
        $subgroupFailKey = "subgroup_" . $subgroupId . "_fail_count";

        // 在redis中删除send list
        $redis->del($subgroupKey);
        $groupLen = $redis->llen($groupKey);
        if (0 == $groupLen) {
            $redis->zrem($mainKey, [$groupKey]);
        }
        // 调整redis中父分组、子分组的失败次数
        $redis->hincrby($groupDetailKey, $groupFailKey, $failCountArr['subgroup_fail_count']*(-1));
        $redis->hincrby($groupDetailKey, $subgroupFailKey, $failCountArr['subgroup_fail_count']*(-1));

        // 更新数据库
        $this->sendListModel->updateSendList($groupDetail, $subgroupId, ['state' => 3, 'last_send' => time()]);   // 更新发送失败的sendlist
        $this->subgroupModel->updateSubgroup($groupDetail['group_id'], $subgroupId, ['state' => MailSubgroup::STATE_FAIL]);
        $finishCount = $this->subgroupModel->getFinishCount($groupDetail['group_id']);
        $groupModel->updateStateByCount($groupDetail['group_id'], $finishCount);

        $isExistsRetryGroup = $redis->exists("retry_group_". $groupDetail['group_id']);
        // 如果已完成的数量超过总数并且没有相应的重发分组，那么删除group_detail key
        if ($groupDetail['total_count'] <= $finishCount && 0 == $isExistsRetryGroup) {
            $redis->del("group_detail_" . $groupDetail['group_id']);
        }
    }

    /**
     * 重发进程
     */
    public function actionResend()
    {
        header("Content-type:text/html;charset=utf-8");
        ignore_user_abort(true);     // 忽略用户的中断  包括curl
        set_time_limit(300);

        $redis = $this->redis;
        $retryGroupArr = $redis->zrange("retry_main", 0, -1);
        
        echo "<title>重发</title>";
        $redisLock = new RedisLock($this->redis, "queue_resend", 300);
        if (false === $redisLock->tryLock()) {
            exit("获得redis锁失败，退出重发进程");
        }

        $lock = new FileLock("resend");
        if (false === $lock->tryLock()) {
            exit("获得文件锁失败，退出重发进程");
        }


        foreach ($retryGroupArr as $retryGroup) {
            $groupId = $this->getSuffix($retryGroup);
           
            $groupDetailKey = "group_detail_" . $groupId;   // 发送进程共用group_detail
            $detail = $redis->hgetall($groupDetailKey);
            $groupDetail = $redis->redisArrToCommonArr($detail);
            if (!empty($groupDetail['error_code']) && in_array($groupDetail['error_code'],$this->config['errorCode']) && !empty($groupDetail['group_last_send']) && time() - $groupDetail['group_last_send'] <  $this->config['sendSpacingTime']) {
              
                $this->writeLog($groupId, "[{$groupId}]上一次发送错误码为{$groupDetail['error_code']}, 并且未达到重发的间隔时间,上次重发/发送时间为" . date("Y-m-d H:i:s",$groupDetail['group_last_send']));
                echo "超限：".$groupId."<br>";
            } else {
                echo "正常发送：" .$groupId."<br>";
            }
           
            $groupKey = "retry_group_" . $groupId;

            $len = $redis->llen($groupKey);
            $this->writeLog($groupId, "开始重发group_id[{$groupId}], 重发子分组数[{$len}]");
            for ($i = 0; $i < $len; $i++) {
                $subgroupKey = $redis->lpop($groupKey);
                $this->writeLog($groupId, "重发: 取出子分组[{$subgroupKey}]");
                if (empty($subgroupKey)) {
                    break;
                }

                try {
                    $this->sendRetrySubgroup("resend", $groupDetail, $subgroupKey);
                } catch (\Exception $e) {
                    $this->writeLog($groupId, "重发：子分组[{$subgroupKey}]重发异常[" . $e->getMessage() . "]");

                    // 如果还存在子分组的sendList，那么就说明没有发送成功。此时需要把subgroupKey补充回groupKey列表
                    if ($redis->exists($subgroupKey)) {
                        $redis->rpush($groupKey, $subgroupKey);
                    }
                }
                sleep(1);       // 休眠1秒，防止进程挂掉

                if (time() - $this->startTime > $this->config['maxResendProcessTime']) {
                    exit("重发进程超过最大运行时间，现退出进程");
                }
            }
        }

        exit("重发进程运行完毕.");
    }


    /**
     * 检测进程（检测暂停分组）
     */
    public function actionCheck()
    {
        ignore_user_abort(true);     // 忽略用户的中断
        set_time_limit(300);

        $maxSuspendLen = $this->config['maxSuspendLen'];    // 最大暂停队列长度
        $maxSuspendTime = $this->config['maxSuspendTime'];   // 最大暂停时间（单位：秒）

        echo "<title>检测进程</title>";

        $redisLock = new RedisLock($this->redis, "queue_check", 300);
        if (false === $redisLock->tryLock()) {
            $msg = "获得redis锁失败，退出检测进程";
            $this->writeProcedureLog($msg);
            exit($msg);
        }

        $lock = new FileLock("check");
        if (false === $lock->tryLock()) {
            exit("获得文件锁失败，退出检测进程");
        }

        $redis = $this->redis;
        $groupRedisArr = $redis->zrange("main", 0, -1, true);
        $groupArr = $redis->redisArrToCommonArr($groupRedisArr);
        $suspendNum = 0;        // 暂停分组的个数
        foreach ($groupArr as $groupKey => $score) {
            $groupId = $this->getSuffix($groupKey);
            $detail = $redis->hgetall("group_detail_" . $groupId);
            $groupDetail = $redis->redisArrToCommonArr($detail);
            if ($this->config['stopPriority']  != $score) {
                continue;
            }

            // 分组为暂停删除的分组
            if (time() - $groupDetail['suspend_time'] >= $maxSuspendTime) {      // 超过最大暂停时间，清理暂停分组数据
                $suspendNum++;
                $this->dealSuspendGroup($groupId);
                continue;
            }

            $len = $redis->llen($groupKey);//
            if ($len >= $maxSuspendLen) {      // 超过最大暂停时间，清理暂停分组数据
                $suspendNum++;
                $this->dealSuspendGroup($groupId);
                continue;
            }
        }
        echo "检测结束：暂停分组[{$suspendNum}]个";

    }

    /**
     * 处理暂停分组
     * (1)清理redis中暂停分组数据
     * (2)更新数据库中子分组的状态为暂停
     * @param $groupId
     */
    private function dealSuspendGroup($groupId)
    {
        $groupKey = "group_" . $groupId;
        $redis = $this->redis;
        $subgroupModel = $this->subgroupModel;
        $subgroupIdArr = [];

        // 删除redis子分组数据
        $subgroupKeyArr = $redis->lrange($groupKey, 0 , -1);
        $redis->del($groupKey);         // 清除父分组队列，
        if (!empty($subgroupKeyArr)) {
            foreach ($subgroupKeyArr AS $v) {
                $subgroupIdArr[] = $this->getSuffix($v);
            }
            $redis->del($subgroupKeyArr);   // 清除子分组下的邮件元素集合
        }

        $redis->zrem("main", $groupKey);

        // 更新数据库
        if (empty($subgroupIdArr)) {
            return;
        }
        $subgroupModel->updateSuspend($groupId, $subgroupIdArr);
    }

    /**
     * 后台页面暂停分组
     * note: 1、此处暂停分组不会修改mail_subgroup表中子分组状态。待检测进程检测到子分组暂停达到阈值才会清除reids队列，
     *          才会修改mail_subgroup表中子分组的状态，意味着真正将分组暂停，
     * 2、当mail_group的state为启动，而mail_subgroup表中为暂停的子分组，生产进程会重新将该子分组加入到redis内存队列
     */
    public function actionSuspend()
    {
       
        $groupId = yii::$app->request->get("groupId");
        $type = yii::$app->request->get("type");
        if (empty($groupId)) {
            echo json_encode(["code" => 10001, "msg" => "分组id不能为空"]);
            return ;
        }
        $session = Yii::$app->session;
        if(!isset($session['admin_info']) || !$session['admin_info']) {
            echo json_encode(["code" => 10005, "msg" => "用户信息失效"]);
            return ;
        }
        
        //判断当前分组是否为已发送状态，如果是则退出
        if(isset($type) && $type == 'ab_test') {
            
            // 更新数据库
            $this->groupModel->updateGroup($groupId, ["state" => MailGroup::STATE_STOP]);
            
            $testGroups = $this->groupModel->getTestChildsGroupListByGroupId($groupId);
            foreach($testGroups as $testGroup) {
                $curGroupInfo = $this->groupModel->getGroupById($testGroup['group_id']); 
                if($curGroupInfo['state'] == MailGroup::STATE_FINISH) {
                    continue;
                }
                $redis = $this->redis;
                $groupModel = $this->groupModel;
                $groupKey = "group_" . $testGroup['group_id'];
                $groupDetailKey = "group_detail_" . $testGroup['group_id'];

                try {
                    // 更新redis
                    $result = $redis->hmset($groupDetailKey, "state", MailGroup::STATE_STOP, "suspend_time", time(),
                        "priority", $this->config['stopPriority']);
                    if (false === $result) {
                        continue;
                    }
                    $result = $redis->zadd("main", [$this->config['stopPriority'], $groupKey]);
                    if (false === $result) {
                        continue;
                    }

                    // 更新数据库
                    $groupModel->updateGroup($testGroup['group_id'], ["state" => MailGroup::STATE_STOP]);
                    
                    if(isset($session['admin_info']['id']) && isset($session['admin_info']['username'])) {
                        //添加操作日志
                        $log = array(
                            'user_id'  => $session['admin_info']['id'],
                            'user_name'  => $session['admin_info']['username'],
                            'group_id'  => $testGroup['group_id'],
                            'operation' => '暂停'
                        );
                        $logObj = new GroupOperationLog();
                        $logObj->addLog($log);
                    //添加操作日志结束
                    }
                    
                } catch (\Exception $e) {
                    $this->returnJson(10002, $e->getMessage());
                }
                
            }
            
        } else {
            $curGroupInfo = $this->groupModel->getGroupById($groupId); 
            if($curGroupInfo['state'] == MailGroup::STATE_FINISH) {
                echo json_encode(["code" => 10003, "msg" => "分组id已发送完成！"]);
                return ;
            }
        
            $redis = $this->redis;
            $groupModel = $this->groupModel;
            $groupKey = "group_" . $groupId;
            $groupDetailKey = "group_detail_" . $groupId;

            try {
                // 更新redis
                $result = $redis->hmset($groupDetailKey, "state", MailGroup::STATE_STOP, "suspend_time", time(),
                    "priority", $this->config['stopPriority']);
                if (false === $result) {
                    throw new \Exception("redis:修改group_detail失败");
                }
                $result = $redis->zadd("main", [$this->config['stopPriority'], $groupKey]);
                if (false === $result) {
                    throw new \Exception("redis:修改main失败");
                }

                // 更新数据库
                $groupModel->updateGroup($groupId, ["state" => MailGroup::STATE_STOP]);
                
                if(isset($session['admin_info']['id']) && isset($session['admin_info']['username'])) {
                    //添加操作日志
                    $log = array(
                        'user_id'  => $session['admin_info']['id'],
                        'user_name'  => $session['admin_info']['username'],
                        'group_id'  => $groupId,
                        'operation' => '暂停'
                    );
                    $logObj = new GroupOperationLog();
                    $logObj->addLog($log);
                    //添加操作日志结束
                }
            } catch (\Exception $e) {
                $this->returnJson(10002, $e->getMessage());
            }
        }
        $this->returnJson(0, "success");
    }

    /**
     * 启动分组
     */
    public function actionStart()
    {
        $groupId = yii::$app->request->get("groupId");
        $type = yii::$app->request->get("type");
        if (empty($groupId)) {
            echo json_encode(["code" => 10001, "msg" => "分组id不能为空"]);
            return ;
        }
        
         $session = Yii::$app->session;
        if(!isset($session['admin_info']) || !$session['admin_info']) {
            echo json_encode(["code" => 10005, "msg" => "用户信息失效"]);
            return ;
        }
        
        if(isset($type) && $type == 'ab_test') {
            
            
            // 更新数据库
            $this->groupModel->updateGroup($groupId, ["state" => MailGroup::STATE_START, "boot_time" => time()]);
            
            $testGroups = $this->groupModel->getTestChildsGroupListByGroupId($groupId);
            foreach($testGroups as $testGroup) {
                $curGroupInfo = $this->groupModel->getGroupById($testGroup['group_id']); 
                if($curGroupInfo['state'] == MailGroup::STATE_FINISH) {
                    continue;
                }

                $redis = $this->redis;
                $groupModel = $this->groupModel;
                $groupKey = "group_" . $testGroup['group_id'];

                try {
                    $info = $groupModel->getGroupById($testGroup['group_id']);
                    if (empty($info)) {
                        continue;
                    }
                    // 更新redis
                    $info['state'] = MailGroup::STATE_START;
                    $info['suspend_time'] = 0;
                    $this->addGroupDetail($info);

                    $result = $redis->zadd("main", [$info['priority'], $groupKey]);
                    if (false === $result) {
                        continue;
                    }

                    // 更新数据库
                    $groupModel->updateGroup($testGroup['group_id'], ["state" => MailGroup::STATE_START, "boot_time" => time()]);
                    
                    
                    //添加操作日志
                   
                   if(isset($session['admin_info']['id']) && isset($session['admin_info']['username'])) {
                       $log = array(
                           'user_id'  => $session['admin_info']['id'],
                           'user_name'  => $session['admin_info']['username'],
                           'group_id'  => $testGroup['group_id'],
                           'operation' => '启动'
                       );
                       $logObj = new GroupOperationLog();
                       $logObj->addLog($log);
                       //添加操作日志结束
                    }

                } catch (\Exception $e) {
                    $this->returnJson(10002, $e->getMessage());
                }
            }
        } else {
            //判断当前分组是否为已发送状态，如果是则退出
            $curGroupInfo = $this->groupModel->getGroupById($groupId); 
            if($curGroupInfo['state'] == MailGroup::STATE_FINISH) {
                echo json_encode(["code" => 10003, "msg" => "分组id已发送完成！"]);
                return ;
            }

            $redis = $this->redis;
            $groupModel = $this->groupModel;
            $groupKey = "group_" . $groupId;

            try {
                $info = $groupModel->getGroupById($groupId);
                if (empty($info)) {
                    throw new \Exception("分组{$groupId}不存在");
                }
                // 更新redis
                $info['state'] = MailGroup::STATE_START;
                $info['suspend_time'] = 0;
                $this->addGroupDetail($info);

                $result = $redis->zadd("main", [$info['priority'], $groupKey]);
                if (false === $result) {
                    throw new \Exception("redis:修改main失败");
                }

                // 更新数据库
                $groupModel->updateGroup($groupId, ["state" => MailGroup::STATE_START, "boot_time" => time()]);
                 //添加操作日志
                
                if(isset($session['admin_info']['id']) && isset($session['admin_info']['username'])) {
                    $log = array(
                        'user_id'  => $session['admin_info']['id'],
                        'user_name'  => $session['admin_info']['username'],
                        'group_id'  => $groupId,
                        'operation' => '启动'
                    );
                    $logObj = new GroupOperationLog();
                    $logObj->addLog($log);
                    //添加操作日志结束
                }

            } catch (\Exception $e) {
                $this->returnJson(10002, $e->getMessage());
            }
        }
        $this->returnJson(0, "success");
    }

    /**
     * 删除分组
     */
    public function actionDelete()
    {
        $groupIdArrTmp = yii::$app->request->get("groupId");
        $type = yii::$app->request->get("type");
        
        $session = Yii::$app->session;
        if(!isset($session['admin_info']) || !$session['admin_info']) {
            echo json_encode(["code" => 10005, "msg" => "用户信息失效"]);
            return ;
        }
        
        if(isset($type) && $type == 'ab_test') {
            $testGroups = $this->groupModel->getTestChildsGroupListByGroupId($groupIdArrTmp);
            // 更新数据库
            $this->groupModel->updateGroup($groupIdArrTmp, ["is_del" => 1]);
            foreach($testGroups as $testGroup) {
                $groupIdArr = [];
                if (!is_array($testGroup['group_id'])) {
                    $groupIdArr[] = $testGroup['group_id'];
                } else {
                    $groupIdArr = $testGroup['group_id'];
                }
                if (empty($groupIdArr)) {
                    continue;
                }
                //判断当前分组是否为已发送状态，如果是则退出
                $curGroupInfo = $this->groupModel->getGroupById($testGroup['group_id']); 
                if($curGroupInfo['state'] == MailGroup::STATE_FINISH) {
                    continue;
                }

                foreach ($groupIdArr as $groupId) {
                    // 删除redis中的分组信息
                    $groupDetailKey = "group_detail_" . $groupId;
                    $this->deleteGroupInRedis($groupId, false);
                    $this->deleteGroupInRedis($groupId, true);
                    $this->redis->del($groupDetailKey);

                    // 更新数据库
                    $this->groupModel->updateGroup($groupId, ["is_del" => 1]);
                }
                 //添加操作日志
                
                if(isset($session['admin_info']['id']) && isset($session['admin_info']['username'])) {
                    $log = array(
                        'user_id'  => $session['admin_info']['id'],
                        'user_name'  => $session['admin_info']['username'],
                        'group_id'  => join(',',$groupIdArr),
                        'operation' => '删除'
                    );
                    $logObj = new GroupOperationLog();
                    $logObj->addLog($log);
                }
            }
        } else {
                if (!is_array($groupIdArrTmp)) {
                    $groupIdArr[] = $groupIdArrTmp;
                } else {
                    $groupIdArr = $groupIdArrTmp;
                }
                if (empty($groupIdArr)) {
                    $this->returnJson(10001, "分组id不能为空");
                }
                //判断当前分组是否为已发送状态，如果是则退出
                $curGroupInfo = $this->groupModel->getGroupById($groupIdArrTmp); 
                if($curGroupInfo['state'] == MailGroup::STATE_FINISH) {
                    $this->returnJson(10003, "分组id已发送完成！");
                }

                foreach ($groupIdArr as $groupId) {
                    // 删除redis中的分组信息
                    $groupDetailKey = "group_detail_" . $groupId;
                    $this->deleteGroupInRedis($groupId, false);
                    $this->deleteGroupInRedis($groupId, true);
                    $this->redis->del($groupDetailKey);

                    // 更新数据库
                    $this->groupModel->updateGroup($groupId, ["is_del" => 1]);
                }
                 //添加操作日志
                
                if(isset($session['admin_info']['id']) && isset($session['admin_info']['username'])) {
                    $log = array(
                        'user_id'  => $session['admin_info']['id'],
                        'user_name'  => $session['admin_info']['username'],
                        'group_id'  => join(',',$groupIdArr),
                        'operation' => '删除'
                    );
                    $logObj = new GroupOperationLog();
                    $logObj->addLog($log);
                }
        }
        //添加操作日志结束
        $this->returnJson(0, "success");
    }

    /**
     * 在redis中删除分组
     * @param $groupId
     * @param $isRetry   bool       是否重发
     */
    private function deleteGroupInRedis($groupId, $isRetry)
    {
        $redis = $this->redis;
        if ($isRetry) {
            $groupKey = "group_" . $groupId;
            $mainKey = "main";
        } else {
            $groupKey = "retry_group_" . $groupId;
            $mainKey = "retry_main";
        }

        // 删除redis子分组数据
        $subgroupKeyArr = $redis->lrange($groupKey, 0 , -1);
        $redis->del($groupKey);             // 清除父分组队列，
        if (!empty($subgroupKeyArr)) {
            foreach ($subgroupKeyArr AS $v) {
                $subgroupIdArr[] = $this->getSuffix($v);
            }
            $redis->del($subgroupKeyArr);       // 清除子分组下的邮件元素集合
        }
        $redis->zrem($mainKey, $groupKey);
    }

    /**
     * 更新分组的worker数
     */
    public function actionWorkerNum()
    {
        $groupId = yii::$app->request->get("groupId");
        $workerNum = yii::$app->request->get("workerNum");

        if (empty($groupId) || !is_numeric($workerNum)) {
            $this->returnJson(10001, "数据不完整");
        }

        $this->groupModel->updateGroup($groupId, ["worker_num" => $workerNum]);
        $this->returnJson(0, "success");
    }

    /**
     * 返回json数据
     * @param $code             编码
     * @param $msg              信息（对编码的解释）
     * @param string $data      数据信息
     */
    public function returnJson($code, $msg, $data = '')
    {
        echo json_encode(['code' => $code, 'msg' => $msg, 'data' => $data]);
        exit();
    }

    /**
     * 根据groupId写日志，一个分组对应一个日志文件
     * @param $groupId
     * @param $msg
     */
    private function writeLog($groupId, $msg)
    {
        $dir = yii::$app->getRuntimePath() . "/logs/queue/" . yii::$app->bjTime->date("Ymd");
        if (!is_dir($dir)) {
            mkdir($dir, 0777, true);
        }
        $filePath = $dir . "/queue_{$groupId}.log";
        $curDate = yii::$app->bjTime->date("Y-m-d H:i:s");
        $msg = "[{$curDate}] " . $msg . PHP_EOL;
        file_put_contents($filePath, $msg, FILE_APPEND);
    }

    /**
     * 写生产进程日志
     * @param $msg
     */
    private function writeProcedureLog($msg)
    {
        $this->writeLog("procedure", $msg);
    }
}

